package com.atguigu.gmall.realtime.app.dws;

import com.atguigu.gmall.realtime.app.func.KeywordUDTF;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author Felix
 * @date 2023/10/10
 * 搜索关键词聚合统计
 * 需要启动的进程
 *      zk、kafka、flume、doris、DwdTrafficBaseLogSplit、DwsTrafficSourceKeywordPageViewWindow
 */
public class DwsTrafficSourceKeywordPageViewWindow {
    public static void main(String[] args) {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //1.3 指定表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //1.4 注册函数
        tableEnv.createTemporarySystemFunction("ik_analyze", KeywordUDTF.class);
        //TODO 2.检查点相关的设置(略)
        env.enableCheckpointing(5000L);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
        //TODO 3.从kafka的页面日志出题中读取数据 创建动态表 并指定Watermark以及提取事件时间字段
        String topic = "dwd_traffic_page_log";
        String groupId = "dws_traffic_keyword_group";
        tableEnv.executeSql("create table page_log(\n" +
            "    common map<string,string>,\n" +
            "    page map<string,string>,\n" +
            "    ts bigint,\n" +
            "    rowtime as TO_TIMESTAMP(FROM_UNIXTIME(ts/1000)),\n" +
            "    WATERMARK FOR rowtime AS rowtime \t\n" +
            ")" + MyKafkaUtil.getKafkaDDL(topic,groupId));
        // tableEnv.executeSql("select * from page_log").print();
        //TODO 4.过滤出搜索行为
        Table searchTable = tableEnv.sqlQuery("select \n" +
            "    page['item'] fullword,\n" +
            "    rowtime\n" +
            "from page_log where page['last_page_id']='search' \n" +
            "   and page['item_type']='keyword' and page['item'] is not null");
        tableEnv.createTemporaryView("search_table",searchTable);
        // tableEnv.executeSql("select * from search_table").print();
        //TODO 5.使用自定义的UDTF函数进行分词 并和原表字段进行关联
        Table joinedTable = tableEnv.sqlQuery("SELECT \n" +
            "    keyword,rowtime\n" +
            "FROM search_table,\n" +
            "LATERAL TABLE(ik_analyze(fullword)) t(keyword)");
        tableEnv.createTemporaryView("joined_table",joinedTable);
        // tableEnv.executeSql("select * from joined_table").print();
        //TODO 6.分组、开窗、聚合计算
        Table reduceTable = tableEnv.sqlQuery("select\n" +
            " DATE_FORMAT(TUMBLE_START(rowtime, INTERVAL '10' second), 'yyyy-MM-dd HH:mm:ss') stt,\n" +
            " DATE_FORMAT(TUMBLE_END(rowtime, INTERVAL '10' second), 'yyyy-MM-dd HH:mm:ss') edt,\n" +
            " DATE_FORMAT(TUMBLE_START(rowtime, INTERVAL '10' second), 'yyyy-MM-dd') cur_date,\n" +
            " keyword,\n" +
            " count(*) keyword_count\n" +
            "from joined_table group by TUMBLE(rowtime, INTERVAL '10' second), keyword");
        tableEnv.createTemporaryView("reduce_table",reduceTable);
        // tableEnv.executeSql("select * from reduce_table").print();
        //TODO 7.将聚合的结果写到Doris表中
        tableEnv.executeSql("create table doris_t(\n" +
            "\tstt string,\n" +
            "\tedt string,\n" +
            "\tcur_date string,\n" +
            "\tkeyword string,\n" +
            "\tkeyword_count bigint\n" +
            ")WITH (\n" +
            "      'connector' = 'doris',\n" +
            "      'fenodes' = '"+ GmallConfig.DORIS_FE +"',\n" +
            "      'table.identifier' = '"+GmallConfig.DORIS_DB+".dws_traffic_source_keyword_page_view_window',\n" +
            "      'username' = 'root',\n" +
            "      'password' = 'aaaaaa',\n" +
            "      'sink.enable-2pc' = 'false'\n" +
            ")");
        tableEnv.executeSql("insert into doris_t select * from reduce_table");
    }
}
